package com.huawei.dli.flink.sye.source;

import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class RandomSource implements SourceFunction<String> {
  private static final Logger logger = LoggerFactory.getLogger(RandomSource.class);

  private long sendInterval;
  private volatile boolean isRunning = true;

  public RandomSource(long sendInterval) {
    this.sendInterval = sendInterval;
  }

  @Override
  public void run(SourceContext<String> out) throws Exception {
    while (isRunning) {
      String value = String.valueOf(Math.floor(Math.random() * 100));
      out.collect(value);
      logger.info("Send value: " + value);
      Thread.sleep(sendInterval);
    }
  }

  @Override
  public void cancel() {
    this.isRunning = false;
  }
}
